package org.itew.mymapreduce.core;

import org.itew.mymapreduce.io.Readable;

import java.io.IOException;

/**
 * 该类是一个map任务的抽象，它负责接收一个分片并发本片传给用户编写的Map任务
 * @author JiLu
 * 
 */
public class BasicMapperRunner<IK,IV,OK,OV> implements Runnable{
	
	private final Mapper<IK,IV,OK,OV> mapper;
	private final Readable<IK,IV> reader;
	private final DataCollection<OK,OV> outCollecton;
	private final Combiner<OK,OV> combiner;
	
	public BasicMapperRunner(Mapper<IK,IV,OK,OV> mapper,Readable<IK,IV> reader,DataCollection<OK,OV> outCollecton,Combiner<OK, OV> combiner)
	{
		this.mapper = mapper;
		this.reader = reader;
		this.outCollecton = outCollecton;
		this.combiner = combiner;
	}
	
	@Override
	public void run() {
		try {
			
			while(reader.next()){
				mapper.map(reader.getKey(), reader.getValue(), outCollecton);
			}
			
			if(combiner!=null){
				for(OK key:outCollecton.keySet()){
					combiner.combine(key, outCollecton.getValues(key), outCollecton);
				}
			}
			
		} catch (IOException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		}
		
	}
	
}
